04 流式响应

此前示例中的 Agent API 都采用一次性响应：等待 Agent 完成处理后，再将完整结果返回给客户端。对于大模型类应用，这种方式会增加用户等待时间。

流式响应是指服务端在生成数据的同时，将已经生成的部分持续发送给客户端，而不必等待全部内容生成完成。

对于 Agent API，流式响应通常用于改善大模型输出的等待体验。LLM 生成回答需要时间，如果等待完整结果后再返回，客户端可能在数秒内没有任何可展示内容。流式模式下，第一个 token 生成后即可返回并展示。

因此，流式响应适合用于逐字输出、工具调用状态更新、长任务进度反馈等场景。

一、基本原理

HTTP 响应分为两部分：头部和体部。普通响应中，服务器通常会先准备完整的响应体，再一次性发送给客户端。

流式响应中，服务器先发送头部，然后通过**生成器（generator）**分块发送响应体。每yield一次，就会输出一段数据。

python

@app.route("/stream")
def stream():
    def generate():
        for i in range(5):
            yield f"第{i+1}条数据\n"

    return generate(), {"Content-Type": "text/plain"}

客户端会陆续收到：

第1条数据
第2条数据
第3条数据
第4条数据
第5条数据

关键点是yield：每次 yield 输出的数据都会尽快发送给客户端，不需要等待整个函数执行完成。

二、Generator生成器

Python 生成器与普通函数的区别在于：普通函数通过return返回结果，调用结束后函数结束；生成器通过yield返回阶段性结果，下一次调用会从上一次暂停的位置继续执行。

python

def normal_func():
    return 1
    return 2  # 永远不会执行

def generator_func():
    yield 1   # 第一次调用到这里暂停
    yield 2   # 第二次调用从这里继续
    yield 3   # 第三次调用从这里继续

gen = generator_func()
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3

Flask 接收到生成器后，会不断调用next()获取数据，并将每次获取到的数据发送给客户端。

三、SSE格式

Agent API 通常使用**SSE（Server-Sent Events）**承载流式响应。SSE 是一种基于 HTTP 的单向推送协议，格式较简单：

data: 第一条消息\n\n
data: 第二条消息\n\n
data: 第三条消息\n\n

每条消息以data: 开头，以\n\n（两个换行）结尾。

3.1 为什么用SSE

方式	特点	适合场景
普通流式	原始数据流	文件下载、CSV导出
SSE	结构化的事件格式	Agent逐字输出、实时通知
WebSocket	双向通信	聊天室、游戏

SSE 适合 Agent 的逐字输出场景：

基于 HTTP，不需要额外协议
浏览器原生支持（EventSource API）
支持自动重连
格式简单，实现成本低

3.2 SSE数据格式

SSE 支持几种字段，最常用的是data：

data: 普通消息\n\n

event: token\ndata: 你好\n\n

event: done\ndata: [DONE]\n\n

字段	说明
`data`	消息内容，可以有多行（每行都以`data:` 开头）
`event`	事件类型，不指定则默认`message`
`id`	消息ID，客户端可以用`Last-Event-ID`重连
`retry`	重连间隔（毫秒）

四、Flask实现SSE

下面是一个完整的 SSE 示例：

python

from flask import Flask, Response, request

app = Flask(__name__)


@app.route("/stream", methods=["GET"])
def stream():
    def generate():
        # 模拟逐字输出
        text = "你好！我是AI助手，很高兴为你服务。"
        for char in text:
            yield f"data: {char}\n\n"

        # 发送结束标记
        yield "data: [DONE]\n\n"

    return Response(
        generate(),
        content_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginx不缓冲
        },
    )

关键配置如下：

要素	说明
`content_type="text/event-stream"`	告诉客户端这是SSE流
`Cache-Control: no-cache`	禁止缓存，保证实时性
`X-Accel-Buffering: no`	告诉Nginx不要缓冲，否则数据可能累积后再发送
`yield "data: ...\n\n"`	每条消息以`data:` 开头，`\n\n`结尾

4.1 客户端接收（JavaScript）

浏览器端可以使用EventSource接收 SSE：

javascript

const eventSource = new EventSource("/stream");

eventSource.onmessage = function(event) {
    if (event.data === "[DONE]") {
        eventSource.close();
        return;
    }
    // 把收到的字符追加到页面上
    document.getElementById("output").textContent += event.data;
};

eventSource.onerror = function() {
    console.log("连接出错");
    eventSource.close();
};

4.2 客户端接收（fetch + ReadableStream）

如果需要通过 POST 发送用户消息，EventSource不适用，因为它只支持 GET。此时可以使用fetch配合ReadableStream：

javascript

async function chat(message) {
    const response = await fetch("/chat", {
        method: "POST",
        headers: {"Content-Type": "application/json"},
        body: JSON.stringify({message: message}),
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();

    while (true) {
        const {done, value} = await reader.read();
        if (done) break;

        const text = decoder.decode(value);
        // 解析SSE格式
        const lines = text.split("\n");
        for (const line of lines) {
            if (line.startsWith("data: ")) {
                const data = line.slice(6);
                if (data === "[DONE]") return;
                document.getElementById("output").textContent += data;
            }
        }
    }
}

五、流式Agent API

将 SSE 与 Agent 接口结合，可以实现流式对话接口：

python

from flask import Flask, Response, request, jsonify
import time

app = Flask(__name__)


@app.route("/chat", methods=["POST"])
def chat():
    """流式对话接口"""
    data = request.get_json()
    if not data or "message" not in data:
        return jsonify({"error": "缺少message字段"}), 400

    message = data["message"]

    def generate():
        # 这里后面会替换成真正的Agent流式调用
        reply = f"你说的是「{message}」，这是一个模拟的流式回复。"
        for char in reply:
            yield f"data: {char}\n\n"
            time.sleep(0.05)  # 模拟LLM生成延迟

        yield "data: [DONE]\n\n"

    return Response(
        generate(),
        content_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

测试：

bash

curl -N -X POST http://127.0.0.1:5000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "你好"}'

-N参数用于禁用 curl 缓冲，使终端能够实时显示流式输出。

六、stream_with_context

在生成器函数内部，不建议直接访问request对象。Flask 开始流式响应后，请求上下文可能已经结束，此时访问request会导致运行时错误。

python

# ❌ 错误：生成器内访问request会报错
@app.route("/stream")
def stream():
    def generate():
        name = request.args.get("name")  # RuntimeError!
        yield f"Hello {name}"
    return generate()

如果需要在流式过程中保留请求上下文，可以使用stream_with_context包装生成器：

python

from flask import stream_with_context

# ✅ 正确：用stream_with_context保持上下文
@app.route("/stream")
def stream():
    name = request.args.get("name", "World")

    def generate():
        # 这里可以安全地使用在generate之前获取的变量
        yield f"Hello {name}！\n\n"
        for i in range(3):
            yield f"data: 第{i+1}条消息\n\n"

    return Response(
        stream_with_context(generate()),
        content_type="text/event-stream",
    )

推荐做法是：在生成器外部获取请求数据，在生成器内部只使用已经保存的变量：

python

@app.route("/chat", methods=["POST"])
def chat():
    # 在生成器外部获取请求数据
    data = request.get_json()
    message = data.get("message", "")

    def generate():
        # 在生成器内部使用已经获取的变量，不要访问request
        for char in f"收到: {message}":
            yield f"data: {char}\n\n"
        yield "data: [DONE]\n\n"

    return Response(
        stream_with_context(generate()),
        content_type="text/event-stream",
    )

七、JSON格式的SSE

实际项目中，SSE 消息通常不只包含纯文本，还可能包含 token 类型、工具调用状态、结束标记等结构化信息。可以将每条消息包装为 JSON：

python

import json

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    message = data.get("message", "")

    def generate():
        # 每条消息是JSON格式
        reply = f"收到: {message}"
        for char in reply:
            chunk = {"content": char, "type": "token"}
            yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"

        # 结束标记也是JSON
        done = {"type": "done"}
        yield f"data: {json.dumps(done)}\n\n"

    return Response(
        stream_with_context(generate()),
        content_type="text/event-stream",
    )

客户端接收到的数据如下：

data: {"content": "收", "type": "token"}

data: {"content": "到", "type": "token"}

data: {"content": ":", "type": "token"}

...

data: {"type": "done"}

这种方式可以在同一条流中传递多种信息，例如 token 内容、工具调用状态、错误信息等。

八、事件类型

SSE 支持自定义事件类型，可以通过event:字段声明：

python

def generate():
    # 普通token
    yield "event: token\ndata: 你好\n\n"

    # 工具调用开始
    yield "event: tool_start\ndata: {\"tool\": \"search\"}\n\n"

    # 工具调用结果
    yield "event: tool_result\ndata: {\"result\": \"...\"}\n\n"

    # 生成结束
    yield "event: done\ndata: {}\n\n"

客户端可以根据事件类型分别处理：

javascript

const eventSource = new EventSource("/chat");

eventSource.addEventListener("token", (e) => {
    // 显示token
    output.textContent += e.data;
});

eventSource.addEventListener("tool_start", (e) => {
    // 显示"正在搜索..."
    const tool = JSON.parse(e.data);
    status.textContent = `正在调用 ${tool.tool}...`;
});

eventSource.addEventListener("done", (e) => {
    eventSource.close();
});

九、注意事项

9.1 顺序保证

SSE 会保证消息按发送顺序到达客户端，适合需要顺序展示的流式文本。

9.2 连接超时

如果长时间没有数据发送，中间代理（Nginx、CDN 等）可能会认为连接闲置并断开。常见做法是定期发送心跳：

python

import time

def generate():
    last_send = time.time()
    for token in agent_stream():
        yield f"data: {token}\n\n"
        last_send = time.time()

        # 每15秒检查一次，如果没有数据就发心跳
        if time.time() - last_send > 15:
            yield ": keepalive\n\n"  # 注释行，客户端会忽略

9.3 Nginx配置

如果使用 Nginx 反向代理，需要关闭缓冲：

nginx

location /api/ {
    proxy_pass http://127.0.0.1:5000;
    proxy_buffering off;          # 关闭缓冲
    proxy_cache off;              # 关闭缓存
    proxy_read_timeout 300s;      # 超时时间设长一点
    chunked_transfer_encoding on;
}

十、总结

流式响应的核心作用，是让 Agent API 在完整结果生成前持续返回阶段性内容。

生成器：用yield一块一块发送数据
SSE格式：data: ...\n\n，简单可靠
stream_with_context：在生成器中安全使用请求数据
JSON消息：传递结构化的token和事件信息
Nginx配置：关闭proxy_buffering保证实时性

通过流式响应，Agent API 可以支持逐字输出、实时状态展示和长任务进度反馈。

下一篇将介绍配置管理和 App Factory，用于管理不同环境下的 Debug、模型名、API Key、数据库地址等配置。

04 流式响应 ​

一、基本原理 ​

二、Generator生成器 ​

三、SSE格式 ​

3.1 为什么用SSE ​

3.2 SSE数据格式 ​

四、Flask实现SSE ​

4.1 客户端接收（JavaScript） ​

4.2 客户端接收（fetch + ReadableStream） ​

五、流式Agent API ​

六、stream_with_context ​

七、JSON格式的SSE ​

八、事件类型 ​

九、注意事项 ​

9.1 顺序保证 ​

9.2 连接超时 ​

9.3 Nginx配置 ​

十、总结 ​

04 流式响应

一、基本原理

二、Generator生成器

三、SSE格式

3.1 为什么用SSE

3.2 SSE数据格式

四、Flask实现SSE

4.1 客户端接收（JavaScript）

4.2 客户端接收（fetch + ReadableStream）

五、流式Agent API

六、stream_with_context

七、JSON格式的SSE

八、事件类型

九、注意事项

9.1 顺序保证

9.2 连接超时

9.3 Nginx配置

十、总结